package com.bw.utils;

import com.bw.bean.CustomerConsumption;
import com.bw.bean.ShopCustomerMetric;
import com.bw.utils.CustomerMetricSink;
import com.bw.utils.ConsumptionSource;
import com.bw.utils.CustomerMetricService;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class CustomerShopApplication {

    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 为了方便测试，设置并行度为1

        // 2. 读取消费记录数据流
        DataStream<CustomerConsumption> consumptionStream = env.addSource(new ConsumptionSource());

        // 3. 处理数据，计算店铺客户指标
        DataStream<ShopCustomerMetric> metricStream = consumptionStream
                // 按店铺ID分组
                .keyBy(CustomerConsumption::getShopId)
                // 滚动窗口，每10秒计算一次
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                // 聚合计算
                .aggregate(new CustomerMetricService.CustomerMetricAggregate(),
                        new CustomerMetricService.CustomerMetricProcessWindow());

        // 4. 输出结果
        metricStream.addSink(new CustomerMetricSink());

        // 5. 执行任务
        env.execute("Customer Shop Analytics");
    }
}
